package site.luojie.netty.rpc.consumer;

import com.alibaba.fastjson.JSON;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import site.luojie.netty.rpc.common.Server;
import site.luojie.netty.rpc.common.UserService;

import java.util.*;
import java.util.stream.Collectors;


/**
 * 服务消费者启动主程序类
 *
 * @author jie.luo
 * @since 2020-07-06 23:54
 **/
public class RpcConsumerZk {

    public static void main(String[] args) throws Exception {
        ZkClient zkClient = new ZkClient("test_server:2181");
        String root = "/servers";

        List<String> children = zkClient.getChildren(root);
        refresh(zkClient, root, children);

        zkClient.subscribeChildChanges(root, new IZkChildListener() {
            @Override
            public void handleChildChange(String path, List<String> childes) throws Exception {
                refresh(zkClient, root, childes);
            }
        });

        while (true) {

            System.out.println(ServerPool.getServersKey());
            Collection<UserService> servers = ServerPool.getServers();
            for (UserService server : servers) {
                server.sayHello("hello");
            }

            Thread.sleep(2000);

        }
    }

    public static void refresh(ZkClient zkClient, String root, List<String> children) throws InterruptedException {
        List<Server> servers = new ArrayList<>();
        Server server = null;

        for (String child : children) {
            String serverJson = zkClient.readData(root + "/" + child);
            server = JSON.parseObject(serverJson, Server.class);
            server.setKey(child);
            servers.add(server);
        }

        onRefresh(servers);
    }

    public static void onRefresh(List<Server> servers) throws InterruptedException {

        if (servers.size() > ServerPool.getServerNum()) {
            // 服务端上线，向池子中添加服务端连接
            for (Server server : servers) {
                if (!ServerPool.containsServer(server.getKey())) {
                    UserService userService = (UserService) new RpcConsumer().build(server.getHost(), server.getPort()).proxy(UserService.class);
                    ServerPool.putServer(server.getKey(), userService);
                }
            }
        } else if (servers.size() < ServerPool.getServerNum()) {
            // 服务端下线，从池子中剔除无效连接
            // 获取有效连接的key
            Set<String> effectiveServers = servers.stream().map(server -> server.getKey()).collect(Collectors.toSet());
            // 从池子中删除无效连接
            Iterator<String> iterator = ServerPool.getServersKey().iterator();
            while (iterator.hasNext()) {
                String key = iterator.next();
                if (!effectiveServers.contains(key)) {
                    ServerPool.removeServer(key);
                }
            }
        } else {
            // 服务端数量与池子数量一致
            return;
        }


    }
}
